#pragma once

#include "Requestor.hpp"

namespace xu
{
    namespace client
    {
        class TopicManager
        {
        public:
            using SubCallback = std::function<void(const std::string &key, const std::string &msg)>;
            using Ptr = std::shared_ptr<TopicManager>;

            TopicManager(const Requestor::Ptr& requestor):_requestor(requestor)
            {}

            //创建主题
            bool Create(const BaseConnection::Ptr &conn, const std::string& key)
            {
                return CommunalRequest(conn,key,TopicOptype::TOPIC_CREATE);
            }

            //删除主题
            bool Remove(const BaseConnection::Ptr &conn, const std::string& key)
            {
                return  CommunalRequest(conn,key,TopicOptype::TOPIC_REMOVE);
            }

            //订阅主题
            bool Subscribe(const BaseConnection::Ptr &conn, const std::string& key,const SubCallback & scb)
            {
                ScalssbackAppend(key,scb);
                bool ret = CommunalRequest(conn,key,TopicOptype::TOPIC_SUBSCRIBE);
                if(ret == false)
                {
                    ScalssbackRemove(key);
                }
                
                return ret;
            }

            //取消订阅
            bool Cancel(const BaseConnection::Ptr &conn, const std::string& key)
            {
                ScalssbackRemove(key);
                return  CommunalRequest(conn,key,TopicOptype::TOPIC_CANCEL);
            }

            //发布消息
            bool Publish(const BaseConnection::Ptr &conn, const std::string& key,const std::string &msg)
            {
                return  CommunalRequest(conn,key,TopicOptype::TOPIC_PUBLISH,msg);
            }

            //设置回调函数
            void onPublish(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {

                //1.判断是否是发布的消息
                if(msg->Optype() != TopicOptype::TOPIC_PUBLISH)
                {
                    LOG(LogLevel::ERROR)<<"类型错误";
                    return;
                }

                //2.查看响应的回调函数
                SubCallback cb = ScalssbackGet(msg->Key());
                if(cb == nullptr)
                {
                    LOG(LogLevel::ERROR)<<msg->Key()<<":  找不到处理的回调函数";
                    return;
                }

                //3.通过回调函数进行处理
                cb(msg->Key(),msg->Msg());
        
            }
        
        private:
            bool CommunalRequest(const BaseConnection::Ptr &conn, const std::string& key,
                const TopicOptype &optype,  const std::string& msg = "")
            {
                //1.构建请求
                auto req_msg  = MessageFactory::create<TopicRequest>();
                req_msg->SetId(UUID::uuid());
                req_msg->SetKey(key);
                req_msg->SetMType(MType::REQ_TOPIC);
                req_msg->SetOptype(optype);

                //2.判断是否是发布请求
                if(optype == TopicOptype::TOPIC_PUBLISH)
                {
                    req_msg->SetMsg(msg);
                }

                //3.发送请求
                BaseMessage::Ptr rsp_msg;
                bool ret = _requestor->Send(conn,req_msg,rsp_msg);
                if(ret == false)
                {
                    LOG(LogLevel::ERROR)<<"发送请求出错";
                    return false;
                }

                auto topic_rsp_msg = std::dynamic_pointer_cast<TopicResponse>(rsp_msg);
                if(topic_rsp_msg == nullptr)
                {
                    LOG(LogLevel::ERROR)<<"响应向下转型失败";
                    return false;
                }

                if(topic_rsp_msg->Rcode() != RCode::RCODE_OK)
                {
                    LOG(LogLevel::ERROR)<<"出错,具体出错原因：" <<errReason(topic_rsp_msg->Rcode());
                    return false;
                }

                LOG(LogLevel::DEBUG)<<"成功："<<(int)topic_rsp_msg->Rcode();

                return true;
            }

            
            //对_topic_subcalssback增删查
            void ScalssbackAppend(const std::string& key,const SubCallback & scb)
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _topic_subcalssback.insert(std::make_pair(key,scb));
            }

            void ScalssbackRemove(const std::string& key)
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _topic_subcalssback.erase(key);
            }

            SubCallback ScalssbackGet(const std::string& key)
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                auto it = _topic_subcalssback.find(key);
                if(it == _topic_subcalssback.end())
                {
                    return SubCallback();
                }

                return it->second;
            }

        private:
            std::mutex _mutex;
            Requestor::Ptr _requestor;
            std::unordered_map<std::string,SubCallback> _topic_subcalssback;
        };

    }

}